package com.yjc.rabbitMQ.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.yjc.rabbitMQ.utils.RabbitMQUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author shkstart
 * @create 2022-07-30 23:06
 */
public class ConfirmMessage {

    private static final int NUMBER = 1000;

    public static void main(String[] args) throws Exception {
        //1.单个发布确定
//        ConfirmMessage.publicMessageIndividually();//花费时间：509ms
        //2.批量发布确定
//        ConfirmMessage.publicMessageBatch();//花费时间：93ms
        //3.异步发布确定
        ConfirmMessage.publicMessageAsync();//花费时间：29ms ;花费时间：46ms ;花费时间：33ms
    }

    public static void publicMessageIndividually() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMQUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();

        long startTime = System.currentTimeMillis();

        for (int i = 0; i < NUMBER; i++) {

            channel.basicPublish("", QUEUE_NAME, null, (i + "消息").getBytes("UTF-8"));
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("花费时间：" + (endTime - startTime) + "ms");

    }

    public static void publicMessageBatch() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMQUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();

        long startTime = System.currentTimeMillis();

        int batchSize = 100;
        boolean flag = false;
        for (int i = 1; i <= NUMBER; i++) {

            channel.basicPublish("", QUEUE_NAME, null, (i + "消息").getBytes("UTF-8"));
            if (i % batchSize == 0) {
                flag = channel.waitForConfirms();
            }
            if (flag) {
                System.out.println("消息发送成功");
            }
        }

        long endTime = System.currentTimeMillis();
        System.out.println("花费时间：" + (endTime - startTime) + "ms");

    }

    public static void publicMessageAsync() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMQUtils.getChannel();
        ConcurrentSkipListMap<Long, String> concurrentSkipListMap = new ConcurrentSkipListMap<>();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //开启发布确认
        channel.confirmSelect();

        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            System.out.println("确定发布的" + deliveryTag);
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirm = concurrentSkipListMap.headMap(deliveryTag);
                confirm.clear();
            } else {
                String remove = concurrentSkipListMap.remove(deliveryTag);
            }
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("未确定发布的id: " + deliveryTag + "未确定发布的消息：" + concurrentSkipListMap.get(deliveryTag));
        };

        channel.addConfirmListener(ackCallback, nackCallback);

        long startTime = System.currentTimeMillis();

        for (int i = 1; i <= NUMBER; i++) {
            channel.basicPublish("", QUEUE_NAME, null, (i + "消息").getBytes("UTF-8"));
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(), ("消息" + i));
        }


        long endTime = System.currentTimeMillis();
        System.out.println("花费时间：" + (endTime - startTime) + "ms");
    }
}
